
package org.apache.flink.api.common.eventtime;

import org.apache.flink.annotation.Public;

import java.time.Duration;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/*
        水印产生的源代码
 */

@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {

    //默认在构造时 =  Long.MIN_VALUE
    private long maxTimestamp;

    /** The maximum out-of-orderness that this watermark generator assumes. */
    private final long outOfOrdernessMillis;

    /**
     * Creates a new watermark generator with the given out-of-orderness bound.
     *
     * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
     */
    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();

        // 构造maxTimestamp
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
    }

    // ------------------------------------------------------------------------

    // 断点式。 来一条数据，就产生和发送一次水印。数据不来，就不产生不发送。  负责产生和发送水印
    // eventTimestamp: 数据的事件时间属性
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        //作用：保证水印只能单调递增。  水印就是时间，时间就是一条逆流的河。只能递增，不能回退！
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
        //System.out.println(event +"到达后，mts:"+maxTimestamp);

        //产生水印
        Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);
        //System.out.println("即将向下游发送wm:"+watermark.getTimestamp());
        //发送水印
        output.emitWatermark(watermark);
    }

    // 默认 周期性地 负责产生和发送水印  默认200ms
    //  outOfOrdernessMillis = 0
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
       //产生水印
      /*  Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);
        System.out.println(Thread.currentThread().getName()+"  即将向下游发送wm:"+watermark.getTimestamp());
        //发送水印
        output.emitWatermark( watermark);*/
    }
}
